package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TradePaymentBean;
import com.atguigu.gmall.realtime.util.DateFormatUtil;
import com.atguigu.gmall.realtime.util.MyClickHouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AllWindowedStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Author: Felix
 * Date: 2022/6/22
 * Desc: 交易域-支付成功独立用户数以及首次支付成功用户数
 * 需要启动的进程
 *      zk、kafka、maxwell、clickhouse
 *      DwdTradeOrderPreProcess、DwdTradeOrderDetail、DwdTradePayDetailSuc、DwsTradePaymentSucWindow
 */
public class DwsTradePaymentSucWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);

        //TODO 2.检查点相关的设置(略)

        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_pay_detail_suc";
        String groupId = "dws_trade_payment_suc_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.对读取的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        /*{"callback_time":"2022-06-22 15:32:51","payment_type_name":"微信","sku_num":"3","split_payment_amount":"69.0",
        "split_original_amount":"69.0000","payment_type_code":"1102","sku_id":"25","source_type_name":"促销活动",
        "order_detail_id":"383","user_id":"79","province_id":"9","source_type_code":"2404",
        "row_op_ts":"2022-06-22 07:32:30.797Z","sku_name":"金沙河面条 银丝挂面900g*3包 爽滑 细面条 龙须面 速食面",
        "source_id":"1","order_id":"177","ts":"1655883152"}*/
        //jsonObjDS.print(">>>>");
        
        //TODO 5.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<JSONObject> jsonObjWithWatermarkDS = jsonObjDS.assignTimestampsAndWatermarks(
            WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<JSONObject>() {
                        @Override
                        public long extractTimestamp(JSONObject jsonObj, long recordTimestamp) {
                            return jsonObj.getLong("ts") * 1000;
                        }
                    }
                )
        );

        //TODO 6.按照用户的id进行分组
        KeyedStream<JSONObject, String> keyedDS = jsonObjWithWatermarkDS.keyBy(jsonObj -> jsonObj.getString("user_id"));
        
        //TODO 7.使用Flink状态编程，判断是否是独立用户以及是否首次支付成功用户
        SingleOutputStreamOperator<TradePaymentBean> processDS = keyedDS.process(
            new KeyedProcessFunction<String, JSONObject, TradePaymentBean>() {
                private ValueState<String> lastPaySucDateState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    lastPaySucDateState
                        = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastPaySucDateState", String.class));
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<TradePaymentBean> out) throws Exception {
                    //获取上次支付成功日期
                    String lastPaySucDate = lastPaySucDateState.value();
                    //获取当前支付成功日期
                    long ts = jsonObj.getLong("ts") * 1000;
                    String curPaySucDate = DateFormatUtil.toDate(ts);
                    Long uuCt = 0L;
                    Long newCt = 0L;

                    if (StringUtils.isEmpty(lastPaySucDate)) {
                        uuCt = 1L;
                        newCt = 1L;
                        lastPaySucDateState.update(curPaySucDate);
                    } else {
                        if (!lastPaySucDate.equals(curPaySucDate)) {
                            uuCt = 1L;
                            lastPaySucDateState.update(curPaySucDate);
                        }
                    }

                    if (uuCt != 0L || newCt != 0L) {
                        out.collect(new TradePaymentBean(
                            "",
                            "",
                            uuCt,
                            newCt,
                            ts
                        ));
                    }
                }
            }
        );

        //TODO 8.开窗
        AllWindowedStream<TradePaymentBean, TimeWindow> windowDS = processDS.windowAll(TumblingEventTimeWindows.of(
            org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));

        //TODO 9.聚合计算
        SingleOutputStreamOperator<TradePaymentBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradePaymentBean>() {
                @Override
                public TradePaymentBean reduce(TradePaymentBean value1, TradePaymentBean value2) throws Exception {
                    value1.setPaymentSucNewUserCount(value1.getPaymentSucNewUserCount() + value2.getPaymentSucNewUserCount());
                    value1.setPaymentSucUniqueUserCount(value1.getPaymentSucUniqueUserCount() + value2.getPaymentSucUniqueUserCount());
                    return value1;
                }
            },
            new AllWindowFunction<TradePaymentBean, TradePaymentBean, TimeWindow>() {
                @Override
                public void apply(TimeWindow window, Iterable<TradePaymentBean> values, Collector<TradePaymentBean> out) throws Exception {
                    for (TradePaymentBean paymentBean : values) {
                        paymentBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        paymentBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        paymentBean.setTs(System.currentTimeMillis());
                        out.collect(paymentBean);
                    }
                }
            }
        );

        //TODO 10.将计算结果写到ClickHouse中
        reduceDS.print(">>>>");
        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_trade_payment_suc_window values(?,?,?,?,?)"));

        env.execute();
    }
}
